package com.atguigu.realtime.function;

import com.atguigu.realtime.commont.Constant;
import com.atguigu.realtime.util.JdbcUtil;
import com.atguigu.realtime.util.RedisUtil;
import com.atguigu.realtime.util.ThreadPoolUtil;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.async.ResultFuture;
import org.apache.flink.streaming.api.functions.async.RichAsyncFunction;
import redis.clients.jedis.Jedis;

import java.sql.Connection;
import java.sql.SQLException;
import java.text.ParseException;
import java.util.concurrent.ThreadPoolExecutor;

/**
 * @Author lizhenchao@atguigu.cn
 * @Date 2021/11/27 14:26
 */
public abstract class DimAsyncFunction<T> extends RichAsyncFunction<T, T> {
    
    private ThreadPoolExecutor threadPool;
    
    @Override
    public void open(Configuration parameters) throws Exception {
        threadPool = ThreadPoolUtil.getThreadPool();
        
    }
    
    // 核心方法: 异步调用. 每来一条数据, 执行这个方法, 在这里执行异步方法
    // 使用线程池来执行你的逻辑
    @Override
    public void asyncInvoke(T input, ResultFuture<T> resultFuture) throws Exception {
        threadPool.submit(new Runnable() {
            @Override
            public void run() {
                // 实现读取维度数据的业务
                // 获取Phoenix连接, redis连接
                Connection conn = JdbcUtil.getPhoenixConnection(Constant.PHOENIX_URL);
                Jedis redisClient = RedisUtil.getRedisClient();
                
                // 使用连接对象去读取对应的维度信息
                try {
                    addDim(conn, redisClient, input, resultFuture);
                } catch (ParseException e) {
                    e.printStackTrace();
                }
                try {
                    conn.close();
                } catch (SQLException e) {
                    e.printStackTrace();
                }
                redisClient.close(); // 如果从连接池获取的redis客户端, 则是归还给连接池. 如果是是new的方式获取的客户端, 则是关闭客户端
                
            }
        });
    }
    
    protected abstract void addDim(Connection conn, Jedis redisClient, T input, ResultFuture<T> resultFuture) throws ParseException;
    
    // 当数据超时的时候, 触发这个方法
    @Override
    public void timeout(T input,
                        ResultFuture<T> resultFuture) throws Exception {
        throw new RuntimeException("异步超时: 请检测以下各种情况:  由于其他问题导致异步超时.\n" +
                                       " 1. hadoop有没有开启, hdfs能不能用?  hdfs是否退出安全模式\n" +
                                       " 2. redis没有开启. redis开启了, 但是无法连接访问. redis的配置问题.  /etc/redis.conf\n" +
                                       "        大小写\n" +
                                       " 3. hbase没有启动成功. 如果Phoenix起来不来, 一定是hbase出问题\n" +
                                       "        复原:\n" +
                                       "            hdfs /hbase 目录删掉\n" +
                                       "            zk /hbase节点删了\n" +
                                       "            \n" +
                                       " 4. 维度数据出问题\n" +
                                       "        检查6张表是否都在\n" +
                                       "        重点检查 dim_user_info的数据是否4000\n" +
                                       "        每个表 bootstrap\n" +
                                       "        \n" +
                                       "        maxwell的服务起来");
        
    }
}
